LearnGraph.online

Appearance

5.6 Multi-agent ​

本节介绍 LangChain 中的多代理系统设计，实现复杂任务的协作处理。

什么是 Multi-agent？ ​

多代理系统（Multi-agent Systems） 将复杂应用分解为多个专业化的 Agent，通过协作完成任务。

"Instead of relying on a single agent to handle every step, multi-agent architectures allow you to compose smaller, focused agents into a coordinated workflow."

何时使用多代理 ​

场景说明
工具过多单个 Agent 有太多工具，决策质量下降
上下文过大记忆/上下文对单个 Agent 来说过大
需要专业化任务需要不同领域的专家（规划、研究、数学）

两种主要模式 ​

1. 工具调用模式（Tool Calling） ​

主管 Agent 将其他 Agent 作为工具调用：

┌─────────────────────────────────────────┐
│           Supervisor Agent              │
│                                         │
│   "我需要调用研究助手查找信息"            │
└─────────────────────────────────────────┘
         │              │              │
         ▼              ▼              ▼
   ┌──────────┐  ┌──────────┐  ┌──────────┐
   │ Research │  │   Math   │  │  Writer  │
   │  Agent   │  │  Agent   │  │  Agent   │
   └──────────┘  └──────────┘  └──────────┘

特点：

  • ✅ 集中控制流 - 所有路由通过主管
  • ✅ 适合任务编排和结构化工作流
  • ❌ 子 Agent 不直接与用户交互

2. 交接模式（Handoffs） ​

Agent 之间直接转移控制权：

┌──────────┐     ┌──────────┐     ┌──────────┐
│  Agent A │────▶│  Agent B │────▶│  Agent C │
│  (活跃)   │     │  (活跃)   │     │  (活跃)   │
└──────────┘     └──────────┘     └──────────┘
     │                │                │
     └────────────────┴────────────────┘
              用户可以与任何活跃 Agent 交互

特点：

  • ✅ 去中心化控制
  • ✅ 支持复杂的专家对话
  • ✅ 多领域对话，专家轮流接管

工具调用模式实现 ​

基本结构 ​

python
from langchain_core.tools import tool
from langchain.agents import create_agent

# 1. 创建子 Agent
research_agent = create_agent(
    model="gpt-4o",
    tools=[search_web, read_document],
    system_prompt="你是一个研究助手，擅长查找和总结信息。"
)

math_agent = create_agent(
    model="gpt-4o",
    tools=[calculate, solve_equation],
    system_prompt="你是一个数学专家，擅长数学计算和问题求解。"
)

# 2. 将子 Agent 包装为工具
@tool("research_assistant", description="用于研究和查找信息的助手")
def call_research_agent(query: str) -> str:
    """调用研究助手"""
    result = research_agent.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    return result["messages"][-1].content

@tool("math_expert", description="用于数学计算和问题求解的专家")
def call_math_agent(problem: str) -> str:
    """调用数学专家"""
    result = math_agent.invoke({
        "messages": [{"role": "user", "content": problem}]
    })
    return result["messages"][-1].content

# 3. 创建主管 Agent
supervisor = create_agent(
    model="gpt-4o",
    tools=[call_research_agent, call_math_agent],
    system_prompt="""你是一个主管助手，负责协调其他专家完成任务。
    - 使用 research_assistant 进行信息查找
    - 使用 math_expert 进行数学计算
    根据用户需求选择合适的专家。"""
)

完整示例：个人助手 ​

python
from langchain_core.tools import tool
from langchain.agents import create_agent, ToolRuntime
from langgraph.checkpoint.memory import InMemorySaver

# 日历 Agent
@tool
def get_calendar_events(date: str) -> str:
    """获取指定日期的日历事件"""
    return f"{date} 的事件: 10:00 团队会议, 14:00 客户电话"

@tool
def create_calendar_event(title: str, date: str, time: str) -> str:
    """创建日历事件"""
    return f"已创建事件: {title} 于 {date} {time}"

calendar_agent = create_agent(
    model="gpt-4o",
    tools=[get_calendar_events, create_calendar_event],
    system_prompt="你是日历管理助手，帮助用户管理日程。"
)

# 邮件 Agent
@tool
def read_emails(folder: str = "inbox") -> str:
    """读取邮件"""
    return f"{folder} 中有 3 封未读邮件"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    return f"已发送邮件到 {to}"

email_agent = create_agent(
    model="gpt-4o",
    tools=[read_emails, send_email],
    system_prompt="你是邮件管理助手，帮助用户处理邮件。"
)

# 包装为工具
@tool("calendar_assistant", description="管理日历和日程的助手")
def call_calendar(request: str) -> str:
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })
    return result["messages"][-1].content

@tool("email_assistant", description="管理邮件的助手")
def call_email(request: str) -> str:
    result = email_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })
    return result["messages"][-1].content

# 主管 Agent
personal_assistant = create_agent(
    model="gpt-4o",
    tools=[call_calendar, call_email],
    system_prompt="""你是一个个人助手，负责协调日历和邮件管理。
    - 日程相关问题使用 calendar_assistant
    - 邮件相关问题使用 email_assistant
    请友好地帮助用户完成任务。""",
    checkpointer=InMemorySaver(),
)

# 使用
result = personal_assistant.invoke({
    "messages": [{"role": "user", "content": "查看我今天的日程，然后给张三发一封邮件约明天开会"}]
}, config={"configurable": {"thread_id": "session_1"}})

print(result["messages"][-1].content)

上下文工程 ​

多代理系统的关键是决定每个 Agent 接收什么信息：

控制子 Agent 输入 ​

1. 修改提示和工具描述 ​

python
# 通过工具描述引导调用方式
@tool(
    "detailed_research",
    description="""用于深入研究的助手。
    调用时请提供:
    - 具体的研究问题
    - 期望的输出格式
    - 任何相关的背景信息"""
)
def call_research(query: str) -> str:
    # ...

2. 上下文注入 ​

python
from langchain.agents import ToolRuntime

@tool("context_aware_agent")
def call_with_context(query: str, runtime: ToolRuntime) -> str:
    """调用子 Agent 并注入上下文"""
    # 从主 Agent 状态获取上下文
    user_context = runtime.state.get("user_preferences", {})

    # 构建增强的查询
    enhanced_query = f"""
    用户偏好: {user_context}

    任务: {query}
    """

    result = sub_agent.invoke({
        "messages": [{"role": "user", "content": enhanced_query}]
    })
    return result["messages"][-1].content

控制子 Agent 输出 ​

1. 细化提示 ​

python
research_agent = create_agent(
    model="gpt-4o",
    tools=[search_web],
    system_prompt="""你是研究助手。

    输出格式要求:
    1. 摘要（2-3句话）
    2. 关键发现（列表形式）
    3. 信息来源

    不要输出无关内容。"""
)

2. 自定义格式化 ​

python
from langchain.agents import Command

@tool("formatted_research")
def call_and_format(query: str, runtime: ToolRuntime) -> Command:
    """调用并格式化输出"""
    result = research_agent.invoke({
        "messages": [{"role": "user", "content": query}]
    })

    raw_output = result["messages"][-1].content

    # 自定义格式化
    formatted = f"[研究结果]\n{raw_output}\n[/研究结果]"

    # 也可以更新状态
    return Command(
        result=formatted,
        update={"last_research": raw_output}
    )

模式对比 ​

方面工具调用模式交接模式
控制方式集中式去中心化
用户交互仅主管 Agent任何活跃 Agent
复杂专家对话有限强大
实现复杂度较低较高
适用场景任务编排多领域专家系统

高级模式：分层多代理 ​

                    ┌───────────────┐
                    │   CEO Agent   │
                    └───────────────┘
                           │
          ┌────────────────┼────────────────┐
          ▼                ▼                ▼
   ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
   │ Research Mgr │ │ Analysis Mgr │ │  Report Mgr  │
   └──────────────┘ └──────────────┘ └──────────────┘
          │                │                │
     ┌────┴────┐      ┌────┴────┐      ┌────┴────┐
     ▼         ▼      ▼         ▼      ▼         ▼
┌────────┐┌────────┐┌────────┐┌────────┐┌────────┐┌────────┐
│ Web    ││ Doc    ││ Data   ││ Stats  ││ Writer ││ Editor │
│ Search ││ Reader ││ Analyst││ Expert ││        ││        │
└────────┘└────────┘└────────┘└────────┘└────────┘└────────┘
python
# 第一层：专业工作者
web_search_agent = create_agent(...)
doc_reader_agent = create_agent(...)
data_analyst_agent = create_agent(...)

# 第二层：部门经理
research_manager = create_agent(
    model="gpt-4o",
    tools=[call_web_search, call_doc_reader],
    system_prompt="你是研究部门经理..."
)

analysis_manager = create_agent(
    model="gpt-4o",
    tools=[call_data_analyst, call_stats_expert],
    system_prompt="你是分析部门经理..."
)

# 第三层：CEO
ceo_agent = create_agent(
    model="gpt-4o",
    tools=[call_research_manager, call_analysis_manager, call_report_manager],
    system_prompt="你是 CEO，负责协调各部门完成复杂任务..."
)

最佳实践 ​

实践说明
单一职责每个 Agent 专注一个领域
清晰边界明确定义 Agent 的职责范围
最小上下文只传递必要的信息给子 Agent
统一格式子 Agent 输出格式保持一致
错误处理子 Agent 失败时有降级策略
日志追踪记录 Agent 间的调用链路

调试技巧 ​

python
import logging

# 启用调试日志
logging.basicConfig(level=logging.DEBUG)

# 在工具中添加日志
@tool("logged_agent")
def call_with_logging(query: str) -> str:
    print(f"[调用子 Agent] 输入: {query}")

    result = sub_agent.invoke({
        "messages": [{"role": "user", "content": query}]
    })

    output = result["messages"][-1].content
    print(f"[子 Agent 响应] 输出: {output[:100]}...")

    return output

上一节：5.5 Human-in-the-loop

下一节：5.7 Retrieval

基于 MIT 许可证发布。内容版权归作者所有。

Copyright © 2025-present 王几行XING（Bryce Wang）